package com.jason.rxjavatest;

import android.os.Bundle;
import android.os.SystemClock;
import android.support.annotation.Nullable;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.widget.Button;
import android.widget.LinearLayout;
import android.widget.TextView;

import com.jason.rxjavatest.bean.Person;
import com.orhanobut.logger.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import butterknife.Bind;
import butterknife.ButterKnife;
import butterknife.OnClick;
import rx.Observable;
import rx.Subscriber;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.functions.Func3;
import rx.schedulers.Schedulers;

/**
 * 地址 http://www.jianshu.com/p/7816380c8ad5
 * <p>
 * 组合操作符
 * <p>
 * 1.zip操作符：
 * 使用场景    a.一个页面有多个接口，我希望等数据全部返回出来之后一并显示
 * b.用户发图片，可能有多张，我希望用户上传完图片之后一并提示
 * <p>
 * <p>
 * 2.CombineLatest
 * 使用场景 登录页面用户名密码需要同时校验，两个EditText的TextWatcher分别触发不同的Observal，以达到共同校验的效果
 * 3.join
 * 可以简单的理解是，我输入了手机号，然后等待验证码，验证码有效期是1分钟，如果在1分钟内你发射数据，咱俩就可以结合了，1分钟后，你照死发，我都不会理你
 *
 * @Author: Jason
 * @Date: 16/5/12
 * @Time: 16:29
 */

public class RxJavaTest05 extends AppCompatActivity {

    private static final String TAG = RxJavaTest05.class.getSimpleName();

    @Bind(R.id.bt_clear)
    Button mBtClear;
    @Bind(R.id.bt_com)
    Button mBtCom;
    @Bind(R.id.bt_join)
    Button mBtJoin;
    @Bind(R.id.ll_root1)
    LinearLayout mLlRoot1;
    @Bind(R.id.bt_groupjoin)
    Button mGroupjoin;
    @Bind(R.id.ll_root2)
    LinearLayout mLlRoot2;
    @Bind(R.id.bt_merge)
    Button mBtMerge;
    @Bind(R.id.bt_mergedelayerror)
    Button mBtMergedelayerror;

    @OnClick(R.id.bt_mergedelayerror)
    void btMergeDelayErrorClick() {
        useMergeDelayError();
    }

    @OnClick(R.id.bt_merge)
    void btMergeClick() {
        useMerge();
    }


    @OnClick(R.id.bt_groupjoin)
    void btGroupJoin() {

        useGroupJoin();
    }


    @OnClick(R.id.bt_join)
    void btJoinClick() {
        useJoin();
    }

    @OnClick(R.id.bt_com)
    void btComClick() {
        useCombineLatest();
    }

    @OnClick(R.id.bt_clear)
    void clearClick() {
        mTvContent.setText("");
    }

    @Bind(R.id.tv_content)
    TextView mTvContent;

    @Bind(R.id.bt_zip)
    Button mBtZip;

    @OnClick(R.id.bt_zip)
    void zipClick() {
        useZip();
    }

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.rxjava5);
        ButterKnife.bind(this);


    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        ButterKnife.unbind(this);
    }


    private void useCombineLatest() {
        final StringBuilder sb = new StringBuilder();
        Observable.combineLatest(
                Observable.just("one.", "two.", "three"),
                Observable.just("1", "2", "3"),
                new Func2<String, String, String>() {
                    @Override
                    public String call(String s, String s2) {
                        Log.e(TAG, "s------>" + s);
                        Log.e(TAG, "s2------>" + s2);
                        return s + s2;
                    }
                }
        ).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        sb.append(s).append("\n");
                        mTvContent.setText(sb.toString());

                    }
                });
    }


    private void useZip() {

        final StringBuilder sb = new StringBuilder();

        Observable<String> just1 = Observable.just("1", "2", "3");
        Observable<String> just2 = Observable.just("first", "second", "thrid");
        Observable<String> just3 = Observable.just("a", "b", "c");

        Observable.zip(just1, just2, just3, new Func3<String, String, String, String>() {
            @Override
            public String call(String s, String s2, String s3) {
                Log.e(TAG, "s------->" + s);
                Log.e(TAG, "s2------->" + s2);
                Log.e(TAG, "s3------->" + s3);

                return s + s2 + s3;
            }

        }).subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                sb.append(s).append("\n");
                Log.e(TAG, "sb------->" + sb.toString());
                mTvContent.setText(sb.toString());

            }
        });

    }

    private void useJoin() {

        final StringBuilder sb = new StringBuilder();

        Observable o1 = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                subscriber.onNext("one");

            }
        });

        Observable o2 = Observable.create(new Observable.OnSubscribe<String>() {

            @Override
            public void call(Subscriber<? super String> subscriber) {
                try {
                    Thread.sleep(4000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                subscriber.onNext("second");
            }
        });

        Observable o3 = Observable.create(new Observable.OnSubscribe<String>() {

            @Override
            public void call(Subscriber<? super String> subscriber) {
                try {
                    Thread.sleep(6000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                subscriber.onNext("third");
            }
        });

        /**
         * 四个参数是什么意思，o3是目标Observal，
         * 第二个参数是源Observal的有效时间，第三个参数是目标Observal的有效时间，
         * 第四个参数是如果有合并，自行处理合并后的逻辑。这里，第二个参数的有效期只有2秒，
         * 那么算上之前的延时时间，源Observal过期2s之后目标Observal才发射数据，那么这里就不会有结合操作了
         */
        o1.join(o3, new Func1() {
            @Override
            public Object call(Object o) {
                return Observable.timer(2, TimeUnit.SECONDS);
            }
        }, new Func1() {
            @Override
            public Object call(Object o) {
                return Observable.timer(2, TimeUnit.SECONDS);
            }
        }, new Func2() {
            @Override
            public Object call(Object o, Object o2) {
                Log.e(TAG, "o -------->" + o.toString());
                Log.e(TAG, "o2 -------->" + o2.toString());
                return o.toString() + o2.toString();
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber() {

            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Object o) {
                Log.e(TAG, "------onNext------");
                sb.append(o.toString()).append("\n");
                mTvContent.setText(sb.toString());
            }
        });


    }


    /**
     * 跟join类似的还有GroupJoin
     * 大体上是一致的，区别在与他是等到有效期到了，才响应，而且第四个参数入参有区别，
     * join是你对象Observal发射出来的那个值，而GroupJoin是对象Observal发射出的Observal
     */
    private void useGroupJoin() {

        final StringBuilder sb = new StringBuilder();

        Observable o1 = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                subscriber.onNext("one");
                Logger.e("o1----------------one");

            }
        });

        Observable o2 = Observable.create(new Observable.OnSubscribe<String>() {

            @Override
            public void call(Subscriber<? super String> subscriber) {
                try {
                    Thread.sleep(4000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                subscriber.onNext("second");
                Logger.e("o2----------------second");
            }
        });

        Observable o3 = Observable.create(new Observable.OnSubscribe<String>() {

            @Override
            public void call(Subscriber<? super String> subscriber) {
                try {
                    Thread.sleep(6000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                subscriber.onNext("third");
                Logger.e("o3----------------third");
            }
        });


        o1.groupJoin(o3, new Func1() {
            @Override
            public Object call(Object o) {
                return Observable.timer(7, TimeUnit.SECONDS);
            }
        }, new Func1() {
            @Override
            public Object call(Object o) {
                return Observable.timer(2, TimeUnit.SECONDS);
            }
        }, new Func2() {
            @Override
            public Object call(final Object o, Object o2) {
                return ((Observable<String>) o2).flatMap(new Func1<String, Observable<?>>() {
                    @Override
                    public Observable<?> call(String s) {
                        Logger.e("call3---------------s---------->" + s);
                        return Observable.just(s + o.toString());
                    }

                });
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Object o) {
                ((Observable) o).subscribe(new Subscriber() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Object o) {
                        Logger.e("onNext---------------s---------->" + o.toString());
                        mTvContent.setText(o.toString());
                    }
                });
            }
        });

    }


    private void useMerge() {

        Observable.merge(Observable.just("1", "2", "3"), Observable.just("4", "5", "6"))
                .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        mTvContent.setText(s);
                    }
                });


        // 先显示data2的数据，后显示data1的数据
        Observable.merge(getObserListData2(), getObserListData1())
                .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<List<Person>>() {
                    @Override
                    public void call(List<Person> persons) {
                        mTvContent.setText(persons.toString());
                    }
                }, new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {

                    }
                });


    }


    private Observable<List<Person>> getObserListData1() {

        return Observable.create(new Observable.OnSubscribe<List<Person>>() {
            @Override
            public void call(Subscriber<? super List<Person>> subscriber) {

                SystemClock.sleep(3000);

                List<Person> list = new ArrayList<Person>();
                list.add(new Person("jason", "bj"));
                list.add(new Person("jack", "us"));
                list.add(new Person("xingli", "JP"));
                subscriber.onNext(list);
                subscriber.onCompleted();

            }
        });
    }


    private Observable<List<Person>> getObserListData2() {

        return Observable.create(new Observable.OnSubscribe<List<Person>>() {
            @Override
            public void call(Subscriber<? super List<Person>> subscriber) {
                SystemClock.sleep(1000);
                List<Person> list = new ArrayList<Person>();
                list.add(new Person("xixi", "bj"));
                list.add(new Person("haha", "us"));
                list.add(new Person("heihei", "JP"));
                subscriber.onNext(list);
                subscriber.onCompleted();

            }
        });
    }


    /**
     * MergeDelayError，他与Merge的区别在于，Merge在数据发射过程中如果遇到错误，
     * 会立即终止，MergeDelayError则会继续发射，直到数据发射完，才将Error传递给观察者
     */
    private void useMergeDelayError() {

        Observable o5 = Observable.create(new Observable.OnSubscribe<Object>() {
            @Override
            public void call(Subscriber<? super Object> subscriber) {
                for (int i = 0; i < 3; i++) {
                    try {
                        Thread.sleep((i + 1) * 3000);
                        if (i == 1) {
                            subscriber.onError(new Exception("ERROR"));
                        }
                        subscriber.onNext("five");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).subscribeOn(Schedulers.io());

        Observable o3 = Observable.create(new Observable.OnSubscribe<String>() {

            @Override
            public void call(Subscriber<? super String> subscriber) {
                try {
                    Thread.sleep(600);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                subscriber.onNext("third");
                Logger.e("o3----------------third");
            }
        });


        Observable.mergeDelayError(o5, o3).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber() {
            @Override
            public void onCompleted() {
            }
            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }
            @Override
            public void onNext(Object o) {
                Log.e(TAG, "useMergeDelayError----->" + o.toString());
            }
        });


    }
}
